package com.dys.flink.wc.flink;

import com.dys.flink.wc.model.Product;
import com.dys.flink.wc.util.JsonUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;

public class ProductPriceAnalysis {
    private static final Logger logger = LoggerFactory.getLogger(ProductPriceAnalysis.class);

    public static void main(String[] args) throws Exception {
        // 使用配置创建执行环境
        Configuration config = new Configuration();
        config.set(CoreOptions.DEFAULT_PARALLELISM, 1);

        try (StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(config)) {

            // 设置检查点间隔（可选，用于生产环境）
            env.enableCheckpointing(5000);

            // 从socket连接读取数据
            DataStream<String> text = env.socketTextStream("localhost", 9999, "\n");

            // 解析商品数据并按键分组进行价格统计
            DataStream<Tuple2<String, Double>> productPriceSums = text
                    .flatMap(new ProductJsonParser())
                    .keyBy(value -> value.f0) // 按商品ID分组
                    .window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5))) // 使用处理时间窗口
                    .sum(1); // 对价格求和

            // 打印结果
            productPriceSums.print().setParallelism(1);

            logger.info("Starting Flink Product Price Analysis...");
            logger.info("Connecting to localhost:9999 for product data...");

            try {
                env.execute("Product Price Analysis");
            } catch (Exception e) {
                logger.error("Flink job execution failed", e);
                throw e;
            }
        }
    }

    public static class ProductJsonParser implements FlatMapFunction<String, Tuple2<String, Double>> {
        @Override
        public void flatMap(String jsonLine, Collector<Tuple2<String, Double>> out) {
            try {
                String cleanJson = jsonLine.trim();

                // 跳过空行和非JSON数据
                if (!cleanJson.startsWith("{")) {
                    logger.debug("Skipping empty or non-JSON line: {}", cleanJson);
                    return;
                }

                // 使用Jackson解析JSON
                Product product = JsonUtils.fromJson(cleanJson, Product.class);

                if (product != null && product.getId() != null) {
                    String id = product.getId();
                    Double price = product.getPrice();

                    out.collect(new Tuple2<>(id, price));
                    logger.info("Successfully parsed product - ID: {}, Name: {}, Price: {}",
                            id, product.getName(), price);
                } else {
                    logger.warn("Failed to parse product from JSON or missing ID: {}", cleanJson);
                }

            } catch (Exception e) {
                logger.error("Error parsing JSON: {}", jsonLine, e);
            }
        }
    }
}